package com.github.kezhenxu94.playground.java.concurrent.itc;

import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

/**
 * <pre>
 * 
 * 线程间通信 - 信号量
 * 
 * 信号量构造函数 {@code public Semaphore(int permits)} 可以
 * 构造一个最多有 {@code permits} 个许可的信号量，当多个线程在该信
 * 号量上执行 acquire 时，同一时刻最多有 permits 个线程可以获得许
 * 可，其余线程必须等待获得许可的线程使用完信号量，并执行 release 操
 * 作才有机会获得许可。
 * 
 * 值得注意的是，在一个信号量上通过 release 执行释放许可操作并不要求
 * 该线程之前通过 acquire 获取过许可。
 * 
 * 可以使用信号量构建阻塞的数据结构，如阻塞链表。
 * 
 * </pre>
 * 
 * @author kezhenxu
 *
 */
public class StudySemaphore {

	public static void main(String[] args) {
		final Semaphore semaphore = new Semaphore(2);
		final ExecutorService executor = Executors.newCachedThreadPool();
		executor.execute(new Task(semaphore));
		executor.execute(new Task(semaphore));
		executor.execute(new Task(semaphore));
		// 只有其中两个线程能够获得许可，但未必是线程id 为 1 和 2 的线程
		// 这与线程调度有关

		System.out.println("测试BlockingSet");
		final BlockingSet<Integer> set = new BlockingSet<>(5);
		executor.execute(new FastProducer(set));
		executor.execute(new SlowConsumer(set));

		executor.shutdown();
	}

}

// 使用信号量构建阻塞链表，简单起见，只实现最小功能集合
class BlockingSet<T> {
	private Set<T> set;
	private Semaphore semaphore;

	public BlockingSet(int capacity) {
		this.set = Collections.synchronizedSet(new HashSet<>());
		this.semaphore = new Semaphore(capacity);
	}

	public boolean add(T e) {
		boolean wasAdded = false;
		try {
			semaphore.acquire();
			wasAdded = set.add(e);
		} catch (InterruptedException e1) {
			e1.printStackTrace();
		} finally {
			if (!wasAdded)
				semaphore.release();
			// 没有添加成功，释放方法开始占用的一个许可
		}
		return wasAdded;
	}

	public boolean remove(T e) {
		boolean wasRemoved = set.remove(e);
		if (wasRemoved)
			semaphore.release();
		return wasRemoved;
	}
}

class FastProducer implements Runnable {
	private BlockingSet<Integer> set;

	public FastProducer(BlockingSet<Integer> set) {
		this.set = set;
	}

	@Override
	public void run() {
		for (int i = 0; i < 10; i++) {
			if (set.add(i))
				System.out.println("生产 " + i);
		}
	}
}

class SlowConsumer implements Runnable {
	private BlockingSet<Integer> set;

	public SlowConsumer(BlockingSet<Integer> set) {
		this.set = set;
	}

	@Override
	public void run() {
		for (int i = 0; i < 10; i++) {
			try {
				Thread.sleep(3000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			if (set.remove(i))
				System.out.println("消费 " + i);
		}
	}
}

class Task implements Runnable {
	private Semaphore semaphore;

	public Task(Semaphore semaphore) {
		this.semaphore = semaphore;
	}

	@Override
	public void run() {
		try {
			// 获取一个许可
			semaphore.acquire(1);
			System.out.println(Thread.currentThread().getName() + " 获得了许可，继续执行");
			// 模拟获取该许可后的耗时操作
			Thread.sleep(3000);
			semaphore.release();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
}
